/**
 * This file is part of the ID2203 course assignments kit.
 * 
 * Copyright (C) 2009 Royal Institute of Technology (KTH)
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License
 * as published by the Free Software Foundation; either version 2
 * of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */
package assignment1.epfd;

import assignment1.eperfectfailuredetector.EPerfectFailureDetector;
import assignment1.eperfectfailuredetector.RestoreEvent;
import assignment1.eperfectfailuredetector.SuspectEvent;

import java.util.HashSet;
import java.util.Set;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


import se.kth.ict.id2203.flp2p.FairLossPointToPointLink;
import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;
import se.sics.kompics.timer.ScheduleTimeout;
import se.sics.kompics.timer.Timer;

/**
 * The <code>Application0</code> class.
 * 
 * @author Federico Piccinini <fpic@sics.se>
 */
public final class Epfd extends ComponentDefinition {

	Negative<EPerfectFailureDetector> epfd =  provides(EPerfectFailureDetector.class);
	
	Positive<Timer> timer = requires(Timer.class);
	Positive<PerfectPointToPointLink> pp2p = requires (PerfectPointToPointLink.class);

	private static final Logger logger = LoggerFactory
			.getLogger(Epfd.class);

	private int timeDelay;
	private int delta;
	private int period;
	private Set<Address> neighborSet;
	private Set<Address> suspected;
	private Set<Address> alive;
	private Address self;

	/**
	 * Instantiates a new application0.
	 */
	public Epfd() {
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handleHeartbeatTimeout, timer);
		subscribe(handleCheckTimeout, timer);
		subscribe(handleDeliver, pp2p);
	}

	Handler<EpfdInit> handleInit = new Handler<EpfdInit>() {
		public void handle(EpfdInit event) {
			timeDelay = event.getTimeDelay();
			period = timeDelay;
			delta = event.getDelta();
			//logger.info("Received init event, timeDelay: {} delta: {}", timeDelay, delta );
			neighborSet = event.getNeighborSet();
			self = event.getSelf();
			suspected = new HashSet<Address>();
			alive = new HashSet<Address>();
			alive.addAll(neighborSet);
			
			scheduleHeartbeatTimeout(timeDelay);
			scheduleCheckTimeout(period);
		}
	};

	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
		}
	};


	Handler<HeartbeatTimeout> handleHeartbeatTimeout = new Handler<HeartbeatTimeout>() {
		public void handle (HeartbeatTimeout event) {
			//logger.info("Received heartbeat timeout event");
			sendHeartbeat();
			scheduleHeartbeatTimeout(timeDelay);
		}
	};
	
	Handler<CheckTimeout> handleCheckTimeout = new Handler<CheckTimeout>() {
		public void handle (CheckTimeout event) {
			//logger.info("Received check timeout event");
			Set<Address> intersection = new HashSet<Address>();
			intersection.addAll(alive);
			intersection.retainAll(suspected);
			
			if (!intersection.isEmpty()) {
				period += delta;
				logger.info ("Increasing period length to {} ms.", period);
			}
			
			for (Address neighbor : neighborSet) {
				if (!alive.contains(neighbor) && !suspected.contains(neighbor)) {
					suspected.add(neighbor);
					trigger (new SuspectEvent(neighbor, period), epfd);
				} else if (alive.contains(neighbor) && suspected.contains(neighbor)) {
					suspected.remove(neighbor);
					trigger (new RestoreEvent(neighbor, period), epfd);
				}
			}
			alive.clear();
			scheduleCheckTimeout (period);
		}
	};
	
	Handler<HeartbeatMessage> handleDeliver = new Handler<HeartbeatMessage>() {
		public void handle (HeartbeatMessage event) {
			//logger.info("Received heartbeat from process {}", event.getSource() );
			alive.add(event.getSource());
		}
	};


	private void scheduleCheckTimeout (int delay) {
		//logger.info("Scheduling check timeout in {} milliseconds...", delay);
		ScheduleTimeout st = new ScheduleTimeout(delay);
		st.setTimeoutEvent(new CheckTimeout(st));
		trigger(st, timer);
	}
	
	private void scheduleHeartbeatTimeout (int period) {
		//logger.info("Scheduling heartbeat timeout in {} milliseconds...", period);
		ScheduleTimeout st = new ScheduleTimeout(period);
		st.setTimeoutEvent(new HeartbeatTimeout(st));
		trigger(st, timer);
	}

	private void sendHeartbeat () {
		for (Address neighbor : neighborSet) {
			//logger.info("Sending heartbeat to {}", neighbor);
			trigger(new Pp2pSend(neighbor, new HeartbeatMessage(self)),
					pp2p);
		}
	}
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
